/* Copyright (c) 2021 OceanBase and/or its affiliates. All rights reserved.
miniob is licensed under Mulan PSL v2.
You can use this software according to the terms and conditions of the Mulan PSL v2.
You may obtain a copy of Mulan PSL v2 at:
         http://license.coscl.org.cn/MulanPSL2
THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
See the Mulan PSL v2 for more details. */

//
// Created by Wangyunlai on 2023/6/13.
//

#include "sql/executor/create_table_executor.h"

#include "common/log/log.h"
#include "event/session_event.h"
#include "event/sql_event.h"
#include "session/session.h"
#include "sql/stmt/create_table_stmt.h"
#include "storage/db/db.h"
#include "storage/trx/trx.h"

RC CreateTableExecutor::execute(SQLStageEvent *sql_event)
{
  Stmt    *stmt    = sql_event->stmt();
  Session *session = sql_event->session_event()->session();
  ASSERT(stmt->type() == StmtType::CREATE_TABLE,
      "create table executor can not run this command: %d",
      static_cast<int>(stmt->type()));

  CreateTableStmt *create_table_stmt = static_cast<CreateTableStmt *>(stmt);

  const char *table_name = create_table_stmt->table_name().c_str();
  RC rc = session->get_current_db()->create_table(table_name, create_table_stmt->attr_infos(), create_table_stmt->primary_keys(), create_table_stmt->storage_format());
  if (rc != RC::SUCCESS) {
    LOG_WARN("failed to create table. rc=%d:%s", rc, strrc(rc));
    return rc;
  }
  if(create_table_stmt->select_physical_operator().get()!=nullptr) {
    PhysicalOperator* select_physical_operator = create_table_stmt->select_physical_operator().get();
    Trx *trx = session->current_trx();
    trx->start_if_need();
    rc = select_physical_operator->open(trx);
    if (rc != RC::SUCCESS) {
      LOG_WARN("failed to open select physical operator. rc=%d:%s", rc, strrc(rc));
      return rc;
    }
    while(OB_SUCC(select_physical_operator->next())) {
      Tuple* tuple = select_physical_operator->current_tuple();
      if(tuple==nullptr) {
        rc = RC::INTERNAL;
        LOG_WARN("failed to get current tuple from select physical operator. rc=%d:%s", rc, strrc(rc));
        return rc;
      }
      std::vector<Value> values;
      int num = tuple->cell_num();
      for(int i = 0; i < num; i++) {
        Value value;
        rc = tuple->cell_at(i, value);
        if (rc != RC::SUCCESS) {
          LOG_WARN("failed to get value from tuple. rc=%d:%s", rc, strrc(rc));
          return rc;
        }
        values.push_back(value);
      }
      ASSERT(values.size()==create_table_stmt->attr_infos().size(),"value num must match attr num");
      if (rc != RC::SUCCESS) {
        LOG_WARN("failed to get values from tuple. rc=%d:%s", rc, strrc(rc));
        return rc;
      }

      Table* table = session->get_current_db()->find_table(table_name);
      if (table == nullptr) {
        rc = RC::SCHEMA_TABLE_NOT_EXIST;
        LOG_WARN("failed to find table %s. rc=%d:%s", table_name, rc, strrc(rc));
        return rc;
      }
      Record record;
      rc = table->make_record(values.size(),values.data(), record);
      if (rc != RC::SUCCESS) {
        LOG_WARN("failed to make record. rc=%d:%s", rc, strrc(rc));
        return rc;
      }
      rc = trx->insert_record(table, record, values);
      if (rc != RC::SUCCESS) {
        LOG_WARN("failed to insert record. rc=%d:%s", rc, strrc(rc));
        return rc;
      }
    }
    rc = select_physical_operator->close();
    
  }

  return rc;
}